package com.gjy.kafka.k10;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-10-03 11:11:25
 */
public class KafkaProducerApp {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerApp.class);
    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final String TOPIC = "test-topic";
    private static final String GROUP_ID = "producer-group";

    private final KafkaProducerService producerService;
    private final ScheduledExecutorService scheduler;
    private volatile boolean running = true;

    public KafkaProducerApp() {
        Properties props = KafkaConfig.getProducerConfig();
        this.producerService = new KafkaProducerService(props);
        this.scheduler = Executors.newScheduledThreadPool(1);

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }

    /**
     * 启动生产者定时任务
     */
    public void start() {
        logger.info("Kafka生产者应用启动，主题: {}", TOPIC);

        // 初始发送一条消息
        sendMessage("初始消息 - 应用启动");

        // 每5秒发送一条消息
        scheduler.scheduleAtFixedRate(() -> {
            if (running) {
                String message = "定时消息 - " + sdf.format(new Date());
                sendMessage(message);
            }
        }, 5, 5, TimeUnit.SECONDS);

        // 每30秒发送一批消息
        scheduler.scheduleAtFixedRate(() -> {
            if (running) {
                sendBatchMessages();
            }
        }, 30, 30, TimeUnit.SECONDS);

        logger.info("生产者定时任务已启动，每5秒发送一条消息，每30秒发送一批消息");
    }

    /**
     * 发送单条消息
     */
    private void sendMessage(String message) {
        try {
            String key = "key-" + System.currentTimeMillis();

            // 使用同步发送确保消息送达
            RecordMetadata metadata = producerService.sendSync(TOPIC, key, message);

            logger.info("✅ 消息发送成功 - Partition: {}, Offset: {}, Key: {}, Message: {}",
                    metadata.partition(), metadata.offset(), key, message);

        } catch (Exception e) {
            logger.error("❌ 消息发送失败: {}", e.getMessage());
        }
    }

    /**
     * 发送带headers的消息
     */
    private void sendMessageWithHeaders(String message) {
        try {
            String key = "header-key-" + System.currentTimeMillis();
            List<Header> headers =
                    KafkaUtil.createHeaders("source", "producer-app", "timestamp",
                            String.valueOf(System.currentTimeMillis()), "type", "batch");

            producerService.sendWithHeaders(TOPIC, key, message, headers);
            logger.info("📨 带headers消息已发送: {}", message);

        } catch (Exception e) {
            logger.error("❌ 带headers消息发送失败: {}", e.getMessage());
        }
    }

    /**
     * 发送批量消息
     */
    private void sendBatchMessages() {
        try {
            List<String> messages = KafkaUtil.generateTestMessages(3);
            producerService.sendBatch(TOPIC, messages);
            logger.info("📦 批量消息发送完成，数量: {}", messages.size());

        } catch (Exception e) {
            logger.error("❌ 批量消息发送失败: {}", e.getMessage());
        }
    }

    /**
     * 发送异步消息（带回调）
     */
    private void sendAsyncMessage(String message) {
        try {
            String key = "async-key-" + System.currentTimeMillis();
            producerService.sendAsyncWithCallback(TOPIC, key, message);
            logger.info("⚡ 异步消息已发送: {}", message);

        } catch (Exception e) {
            logger.error("❌ 异步消息发送失败: {}", e.getMessage());
        }
    }

    /**
     * 优雅关闭
     */
    public void shutdown() {
        running = false;
        logger.info("正在关闭生产者应用...");

        try {
            scheduler.shutdown();
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }

        producerService.close();
        logger.info("生产者应用已关闭");
    }

    /**
     * 主函数 - 生产者应用入口
     */
    public static void main(String[] args) {
        KafkaProducerApp app = new KafkaProducerApp();

        try {
            // 启动生产者
            app.start();

            // 保持主线程运行，等待用户中断
            logger.info("生产者应用运行中... 按 Ctrl+C 停止");

            // 模拟其他工作或等待
            while (app.running) {
                try {
                    // 每10秒发送一次异步消息
                    Thread.sleep(10000);
                    if (app.running) {
                        String asyncMessage = "异步测试消息 - " + sdf.format(new Date());
                        app.sendAsyncMessage(asyncMessage);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }

        } catch (Exception e) {
            logger.error("生产者应用异常: {}", e.getMessage(), e);
        } finally {
            app.shutdown();
        }
    }

}
